package com.ycy;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class Produces {

  static Logger logger = LoggerFactory.getLogger(Produces.class);

  public static void main(String[] args) {

    Properties properties = new Properties();

    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Conf.IP + ":" + Conf.PORT); //ip
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

    // 这个是线程安全的，可以通过多线程调用
    KafkaProducer kafkaProducer = new KafkaProducer(properties);

//    ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(Conf.TOPIC3, "ycy4", "helloword2");

//    ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(Conf.TOPIC4, "ycy4", "helloword2");



    for (int i = 0; i < 10; i++) {
      try {

        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(Conf.TOPIC4, "helloword" + i);

        Future send = kafkaProducer.send(producerRecord, new Callback() {
          @Override
          public void onCompletion(RecordMetadata metadata, Exception exception) {
            if (exception != null) {
              exception.printStackTrace();
            }
            if (metadata != null) {
              logger.info(String.format("发送成功！！！topic:%s 偏移量:%s 分区:%s", metadata.topic(), metadata.offset(), metadata.partition()));
            }
          }
        });
        try {
          Object o = send.get();
        } catch (InterruptedException | ExecutionException e) {
          e.printStackTrace();
        }
      } catch (Exception e) {
        e.printStackTrace();
      } finally {
//        kafkaProducer.close();
      }

//      try {
//        Thread.sleep(1000);
//      } catch (InterruptedException e) {
//        e.printStackTrace();
//      }

    }

    kafkaProducer.close();

  }

}
